package gspushmq.util;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.google.common.collect.Lists;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;
import org.java_websocket.WebSocket;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Sets;

import ggframework.mongodb.BasicDBObject;
import ggframework.mongodb.DBCollection;
import ggframework.mongodb.DBCursor;
import ggframework.mongodb.DBObject;
import gspushmq.model.Data;
import gspushmq.model.DataMessage;
import gspushmq.model.Message;
import gspushmq.model.MessageType;
import gspushmq.model.User;
import gspushmq.model.VersionEnum;
import gspushmq.v1.produce.job.sys.GGMQRealTimeJob;
import gspushmq.v1.save.SaveSendRecord;
import gspushmq.v1.save.SaveMassRecord;
import gspushmq.v1.save.log.LogBean;
import gspushmq.v1.save.log.LogQueue;
import gspushmq.v1.subscribe.GGSubscribe;
import gspushmq.v1.thread.GGThreadPool;

/**
 * @Description 类系统类消息推送逻辑工具类
 * @Author chengsj
 * @Date 2018/11/21 14:42
 * @Version 1.0
 * @company 上海朝阳永续信息技术有限公司
 * @copyright (c) 2015 SunTime Co'Ltd Inc. All rights reserved.
 **/
public class SysUtil {


    /**
     *
     * @description 获取指定消息类型的生产来源
     *
     * @param mqSubType
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/18 19:25
     *
     */
    public static Set<Object> getSources(String mqSubType){
        DBCollection collection = MongoUtil.getFtAccountBehaviorCollection("cbs_msg_storage_record");
        DBObject query = new BasicDBObject();
        if(StringUtils.isNotBlank(mqSubType)) {
            query.put("mqSubType", mqSubType);
        }
        Set<Object> objects = Sets.newHashSet();
        DBCursor cursor = collection.find(query,new BasicDBObject("_id",0).append("source",1));
        while (cursor.hasNext()){
            DBObject o = cursor.next();
            if (null != o.get("source")) {
                objects.add(o.get("source"));
            }
        }
        cursor.close();
        return objects;
    }


    /**
     *
     * @description 复制同一消息到不同产品线对象中，默认产品线4,901,28
     *
     * @param dataMessage
     * @version 1.0
     * @author 成尚佳
     * @company 上海朝阳永续信息技术有限公司
     * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
     * @date 2019/1/15 14:21
     *
     */
    public static List<Data> dispatherProductLine(Data dataMessage){
        if (dataMessage == null){
            return null;
        }
        List<Data> list = Lists.newArrayList();
        for (Integer productLine : Sets.newHashSet(4, 28, 901) ) {
            Data data1 = deepClone(dataMessage);
            data1.setProductLine(productLine);
            list.add(data1);
        }
        return list;
    }


    /**
     *
     * @description 复制对象
     * @param
     * @version 1.0
     * @author chengsja
     * @date 2018/12/18 10:21
     *
     */
    public static Data deepClone(Data dataMessage){
        DataMessage d1 = new DataMessage();
        d1.setModule(dataMessage.getModule());
        d1.setTarget(dataMessage.getTarget());
        d1.setLinkType(dataMessage.getLinkType());
        d1.setProductLine(dataMessage.getProductLine());
        d1.setTitle(dataMessage.getTitle());
        d1.setUuid(dataMessage.getUuid());
        d1.setContent(dataMessage.getContent());
        d1.setCrowdType(dataMessage.getCrowdType());
        d1.setCrowdValue(dataMessage.getCrowdValue());
        d1.setMqType(dataMessage.getMqType());
        d1.setMqSubType(dataMessage.getMqSubType());
        d1.setExpireDate(dataMessage.getExpireDate());
        d1.setStartDate(dataMessage.getStartDate());
        d1.setUserType(dataMessage.getUserType());
        d1.setLinkUrl(dataMessage.getLinkUrl());
        d1.setSource(dataMessage.getSource());
        return d1;
    }
    /**
     * @param
     * @description 登录消费未过期的消息
     * @version 1.0
     * @author chengsja
     * @date 2018/12/19 17:11
     */
    public static void loginConsumer(Message message, WebSocket ws) {
        Set<DataMessage> pdm = listNoExpiredWaitConsumerMessage(message);
        loginConsumer(message,ws,pdm);
    }

    /**
     *
     * @description 登录消费未过期消息
     * @param
     * @version 1.0
     * @author chengsja
     * @date 2018/12/19 17:28
     *
     */
    public static void loginConsumer(Message message,WebSocket ws,Set<DataMessage> pdm){
        if (pdm != null && pdm.size() > 0) {
            for (DataMessage dm : pdm) {
                if (ws.isOpen()) {
                    sendWebSocket(message, ws, dm);
                }
            }
        }
    }

    /**
     * @param
     * @description 获取未过期的消息集合
     * @version 1.0
     * @author chengsja
     * @date 2018/12/19 17:13
     */
    public static Set<DataMessage> listNoExpiredMessages(Message message) {
        //查生产库
        if (message == null
                //产品线
                || StringUtils.isBlank(message.getProduct_line())
                //账号id
                || StringUtils.isBlank(message.getAccount_id())
                //权限
                || StringUtils.isBlank(message.getPro_type())) {
            return null;
        }
        List<DBObject> orList = new ArrayList<>();
        User user = getOnlineUserByUid(message.getAccount_id(),message.getProduct_line());
        if (user == null){
            return null;
        }
        Integer productLine = Integer.valueOf(message.getProduct_line());
        Long accountId = Long.valueOf(message.getAccount_id());
        String pro = message.getPro_type();
        String accountName = user.getAccountName();
        //1.获取用户终端类型，匹配推送数据为TAG推送
        Integer clientNumer = VersionEnum.getNumber(productLine,Integer.valueOf(pro));
        if (clientNumer != null) {
            DBObject or1 = new BasicDBObject();
            or1.put("crowdType", "TAG");
            or1.put("crowdValue", new BasicDBObject("$regex",String.valueOf(clientNumer)));
            orList.add(or1);
        }
        //2.获取用户名，匹配推送数据为ACCOUNT推送
        if (StringUtils.isNotBlank(accountName)){
            DBObject or2 = new BasicDBObject();
            or2.put("crowdType", "ACCOUNT");
            or2.put("crowdValue", new BasicDBObject("$regex",accountName));
            orList.add(or2);
        }
        //3.获取账号，匹配推送数据为ALAS推送
        if (accountId != null){
            DBObject or3 = new BasicDBObject();
            or3.put("crowdType", "ALIAS");
            or3.put("crowdValue", new BasicDBObject("$regex",String.valueOf(accountId)));
            orList.add(or3);
        }
        DBCollection messRecord = MongoUtil.getFtAccountBehaviorCollection("cbs_msg_storage_record");
        DBObject query = new BasicDBObject();
        query.put("productLine", productLine);
        query.put("mqSubType", message.getMq_sub_type());
        query.put("expireDate", new BasicDBObject("$gte", new Date()));
        query.put("$or",orList);
        DBCursor cursor = messRecord.find(query);
        Set<DataMessage> pdm = Sets.newHashSet();
        while (cursor.hasNext()) {
            DBObject o = cursor.next();
            DataMessage dm = new DataMessage();
            dm.setUuid(o.getString("uuid"));
            dm.setMqType(o.getString("mqType"));
            dm.setMqSubType(o.getString("mqSubType"));
            dm.setTitle(o.getString("title"));
            dm.setContent(o.getString("content"));
            dm.setProductLine(o.getInteger("productLine"));
            dm.setFileType(o.getString("fileType"));
            dm.setFileAddress(o.getString("fileAddress"));
            dm.setLinkType(o.getString("linkType"));
            dm.setLinkUrl(o.getString("linkUrl"));
            dm.setModule(o.getString("module"));
            dm.setTarget(o.getString("target"));
            dm.setCrowdType(o.getString("crowdType"));
            dm.setCrowdValue(o.getString("crowdValue"));
            dm.setStartDate(o.getDate("startDate"));
            dm.setExpireDate(o.getDate("expireDate"));
            pdm.add(dm);
        }
        cursor.close();
        return pdm;
    }

    /**
     *
     * @description 获取未过期待消费的消息
     * @param
     * @version 1.0
     * @author chengsja
     * @date 2018/12/19 17:36
     *
     */
    public static Set<DataMessage> listNoExpiredWaitConsumerMessage(Message message){
        return removeConsumeredMUUID(listNoExpiredMessages(message),message);
    }
    /**
     * @description 获取uuid集合
     * @param messageSet
     * @return set<String>
     * @version 1.0
     * @author chengsja
     * @date 2018/12/14 15:00
     */
    public static Set<String> getUUIDs(Set<DataMessage> messageSet) {
        if (messageSet == null || messageSet.size() == 0) {
            return null;
        }
        Set<String> set = Sets.newHashSet();
        for (DataMessage dataMessage : messageSet) {
            set.add(dataMessage.getUuid());
        }
        return set;
    }


    /**
     * @param pdm 未过期的消息集合
     * @return set<String> 过滤后的uuid集合
     * @description 获取产品线下未过期未消费的消息集合
     * @version 1.0
     * @author chengsja
     * @date 2018/12/14 14:47
     */
    public static Set<DataMessage> removeConsumeredMUUID(Set<DataMessage> pdm,Message message) {
        Set<String> puuids = getUUIDs(pdm);
        if (puuids == null || puuids.size() == 0 || message == null || StringUtils.isBlank(message.getProduct_line())) {
            return null;
        }
        if (puuids.size() > 0) {
            DBCollection messRecord = MongoUtil.getFtAccountBehaviorCollection("cabs_msg_send_record");
            DBCursor cursor = messRecord.find(new BasicDBObject("uuid", new BasicDBObject("$in", puuids))
                    .append("product_line", Integer.valueOf(message.getProduct_line()))
                    .append("account_id",Long.valueOf(message.getAccount_id())), new BasicDBObject("uuid", 1));
            Set<DataMessage> delDm = Sets.newHashSet();
            while (cursor.hasNext()) {
                DBObject o = cursor.next();
                for (DataMessage dm : pdm) {
                    if (o.getString("uuid").equals(dm.getUuid())) {
                        delDm.add(dm);
                    }
                }

            }
            cursor.close();
            //过滤已推送的
            pdm.removeAll(delDm);
        }
        GGLogger.info(SysUtil.class,new StringBuffer("------------>上线消费---------->").append("\n")
        .append("uid=").append(message.getAccount_id()).append("\n")
                .append("productLine=").append(message.getProduct_line()).append("\n")
                .append("mqSubType=").append(message.getMq_sub_type()).append("\n")
                .append("noExpiredUUIDs=").append(JSON.toJSONString(puuids)).append("\n")
                .append("noExpiredNoCousumeredMessage=").append(JSON.toJSONString(pdm)).append("\n")
        );
        return pdm;
    }

    /**
     *
     * @description 获取在线用户信息
     * @param
     * @version 1.0
     * @author chengsja
     * @date 2018/12/19 18:33
     *
     */
    public static User getOnlineUserByUid(String accountId,String productLine){
        if (StringUtils.isBlank(accountId)){
            return null;
        }
        User user = new User();
        user.setProductLine(productLine);
        user.setAccountId(accountId);
        Iterator<Map.Entry<WebSocket,User>> wuitor = GGSubscribe.wstUser.entrySet().iterator();
        while (wuitor.hasNext()){
            Map.Entry<WebSocket,User> userEntry = wuitor.next();
            User u1 = userEntry.getValue();
            if (user.equals(u1)) {
                return userEntry.getValue();
            }
        }
        return null;
    }

    /**
     * @param
     * @description 生产日志公共类
     * @version 1.0
     * @author chengsja
     * @date 2018/12/13 19:45
     */
    public static void logPattern(Class clazz, DataMessage dataMessage) {
        GGLogger.info(clazz, "----消息生产成功:mqType=" + dataMessage.getMqType()
                + "|mqSubType=" + dataMessage.getMqSubType() + "|productLine=" + dataMessage.getProductLine() + "|crowdType=" + dataMessage.getCrowdType()
                + "|crowdValue=" + dataMessage.getCrowdValue());
    }


	/**
	 * 推送消息给客户端,系统类消息日志
	 *
	 * @param ams
	 * @param wst
	 * @param data
	 * @author chengsj
	 * @date 2018年11月25日下午3:47:01
	 */
    public static void sendWebSocket(Message ams, WebSocket wst, Data data) {
        Map<String, Object> result = new HashMap<>();
        result.put("code", 0);
        result.put("request_id", ams.getRequest_id());
        result.put("uuid", data.getUuid());
        result.put("data", new Data(data));
        String sendStr = JSON.toJSONString(result);
        wst.send(sendStr);
        GGLogger.info(GGMQRealTimeJob.class, "-=-=-=消息消费成功：uuid=" + data.getUuid() + "|mq_sub_type=" + data.getMqSubType() + "|uid=" + ams.getAccount_id() + "|productLine=" + ams.getProduct_line());
        SaveSendRecord.save(ams, data);
    }


	/**
	 * 生产消息入生产库 Created by chengsj on 2018/11/23
	 */
    public static void sendToMongo(Data data) {
        //入库
    	SaveMassRecord.save(data);
    }


    /**
     * 获取指定类型的已推送的有效uuid集合
     *
     * @param messageType 消息类型对象，其中mq_sub_type字段不能为空
     * @return Set<String> uuids 一定是非null对象
     * Created by chengsj on 2018/11/23
     */
    public static Set<String> filterAlreadyExistsUUid(MessageType messageType) {
        if (messageType == null){
            return null;
        }
        String mqSubTyp = messageType.getMq_sub_type();
        String mqTyp = messageType.getMq_type();
        Set<String> existsUUidSet = Sets.newHashSet();
        if (messageType == null || StringUtils.isBlank(mqSubTyp) || StringUtils.isBlank(mqTyp)) {
            return null;
        }
        DBCollection messRecord = MongoUtil.getFtAccountBehaviorCollection("cbs_msg_storage_record");
        DBObject query = new BasicDBObject();
        query.put("expireDate", new BasicDBObject("$gte", new Date()));
        query.put("mqSubType", mqSubTyp);
        DBCursor cursor = messRecord.find(query, new BasicDBObject("uuid", 1));
        while (cursor.hasNext()) {
            DBObject o = cursor.next();
            existsUUidSet.add(o.getString("uuid"));
        }
        cursor.close();
        return existsUUidSet;
    }


	/**
	 * 从文件中读取上次最新的时间值,默认是当前时间 Created by chengsj on 2018/11/21
	 */
    public static Date getCacheLastDate(String fileName) throws Exception {
        if (StringUtils.isBlank(fileName)) {
            throw new Exception("获取上次最新时间值存储文件名不能为空!");
        }
        String lastSwapStr = read(fileName);
        if (StringUtils.isBlank(lastSwapStr)) {
            write(fileName, String.valueOf(System.currentTimeMillis()));
            return getCacheLastDate(fileName);
        } else {
            return new Date(Long.valueOf(lastSwapStr));
        }
    }


    /**
     * 缓存时间戳到指定文件中
     * Created by chengsj on 2018/11/21
     */
    public static void cacheTimeSwapToFile(Date cacheDate, String fileName) throws Exception {
        if (cacheDate == null) {
            throw new Exception("缓存时间cacheDate不能为null!");
        }
        if (StringUtils.isBlank(fileName)) {
            throw new Exception("缓存文件名称fileName不能为null!");
        }
        write(fileName, String.valueOf(cacheDate.getTime()));
    }


    /**
     * 读文件
     * Created by chengsj on 2018/11/21
     */
    public static String read(String fileName) throws IOException {
        if (StringUtils.isBlank(fileName)) {
            throw new IOException("执行读操作时，文件名不能为空！");
        }
        File file = new File(fileName);
        if (!file.exists()) {
            if (!file.getParentFile().exists()) {
                file.mkdirs();
            }
            file.createNewFile();
        }
        return FileUtils.readFileToString(file);
    }

    /**
     * 写文件
     * Created by chengsj on 2018/11/21
     */
    public static void write(String fileName, String content) throws IOException {
        if (StringUtils.isBlank(fileName)) {
            throw new IOException("执行写操作时，文件名不能为空！");
        }
        FileUtils.writeStringToFile(new File(fileName), content);
    }
}
